package com.codeduck.kafka.consumer.thread;

import com.codeduck.handler.AbstractMsgHandler;
import com.codeduck.pojo.TopicHandler;
import com.codeduck.utils.SpringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Iterator;
import java.util.Properties;

public class SubscribeMsgRunnable implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(SubscribeMsgRunnable.class);
    private static final String AUTO_COMMIT_PROPS = "enable.auto.commit";
    private static final String TIMEOUT_PROPS = "timeout";

    protected Properties props;
    protected TopicHandler topicHandler;

    public SubscribeMsgRunnable(Properties props, TopicHandler topicHandler) {
        this.props = props;
        this.topicHandler = topicHandler;
    }

    @Override
    public void run() {
        try {
            KafkaConsumer<Object, Object> kafkaConsumer = new KafkaConsumer<>(this.props);
            kafkaConsumer.subscribe(Collections.singletonList(topicHandler.getTopic()));
            while (true) {
                // 从服务端获取消息
                this.pull(kafkaConsumer);
            }
        } catch (Exception e) {
            LOGGER.error("KafkaConsumer Thread 出现异常", e);
        }
    }

    private void pull(KafkaConsumer<Object, Object> kafkaConsumer) {
        ConsumerRecords<Object, Object> records = kafkaConsumer.poll(Long.parseLong((String) props.get(TIMEOUT_PROPS)));
        consume(records);
        // 是否需要手动提交消息
        if ("false".equals(props.get(AUTO_COMMIT_PROPS))) {
            kafkaConsumer.commitSync();
        }
    }

    private void consume(ConsumerRecords<Object, Object> records) {
        AbstractMsgHandler handler = SpringUtils.getBean(topicHandler.getHandler());
        if (topicHandler.isBatchHandler()) {
            LOGGER.info("消息监听器已接收消息：topic[{}]，数量[{}]", topicHandler.getTopic(), records.count());
            handler.batchConsume(records);
        } else {
            Iterator<ConsumerRecord<Object, Object>> iterator = records.iterator();
            while (iterator.hasNext()) {
                ConsumerRecord<Object, Object> record = iterator.next();
                LOGGER.info("消息者已接收消息：topic[{}],partition[{}],offset[{}],timestamp[{}]", record.topic(), record.partition(), record.offset(), record.timestamp());
                handler.consume(record);
            }
        }
    }

}
